/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.dubbo.rpc.cluster.support;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.config.Configuration;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.profiler.ProfilerSwitch;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.InvocationProfilerUtils;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.RpcServiceContext;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.LoadBalance;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ScopeModelUtil;
import org.apache.dubbo.rpc.support.RpcUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_LOADBALANCE;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_RESELECT_COUNT;
import static org.apache.dubbo.common.constants.CommonConstants.ENABLE_CONNECTIVITY_VALIDATION;
import static org.apache.dubbo.common.constants.CommonConstants.LOADBALANCE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.RESELECT_COUNT;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_RESELECT_INVOKERS;
import static org.apache.dubbo.rpc.cluster.Constants.CLUSTER_AVAILABLE_CHECK_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.CLUSTER_STICKY_KEY;
import static org.apache.dubbo.rpc.cluster.Constants.DEFAULT_CLUSTER_AVAILABLE_CHECK;
import static org.apache.dubbo.rpc.cluster.Constants.DEFAULT_CLUSTER_STICKY;

/**
 * AbstractClusterInvoker
 */
public abstract class AbstractClusterInvoker<T> implements ClusterInvoker<T> {

    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractClusterInvoker.class);

    protected Directory<T> directory;

    protected boolean availableCheck;

    private volatile int reselectCount = DEFAULT_RESELECT_COUNT;

    private volatile boolean enableConnectivityValidation = true;

    private final AtomicBoolean destroyed = new AtomicBoolean(false);

    private volatile Invoker<T> stickyInvoker = null;

    public AbstractClusterInvoker() {
    }

    public AbstractClusterInvoker(Directory<T> directory) {
        this(directory, directory.getUrl());
    }

    public AbstractClusterInvoker(Directory<T> directory, URL url) {
        if (directory == null) {
            throw new IllegalArgumentException("service directory == null");
        }

        this.directory = directory;
        //sticky: invoker.isAvailable() should always be checked before using when availablecheck is true.
        this.availableCheck = url.getParameter(CLUSTER_AVAILABLE_CHECK_KEY, DEFAULT_CLUSTER_AVAILABLE_CHECK);
        Configuration configuration = ConfigurationUtils.getGlobalConfiguration(url.getOrDefaultModuleModel());
        this.reselectCount = configuration.getInt(RESELECT_COUNT, DEFAULT_RESELECT_COUNT);
        this.enableConnectivityValidation = configuration.getBoolean(ENABLE_CONNECTIVITY_VALIDATION, true);
    }

    @Override
    public Class<T> getInterface() {
        return getDirectory().getInterface();
    }

    @Override
    public URL getUrl() {
        return getDirectory().getConsumerUrl();
    }

    @Override
    public URL getRegistryUrl() {
        return getDirectory().getUrl();
    }

    @Override
    public boolean isAvailable() {
        Invoker<T> invoker = stickyInvoker;
        if (invoker != null) {
            return invoker.isAvailable();
        }
        return getDirectory().isAvailable();
    }

    @Override
    public Directory<T> getDirectory() {
        return directory;
    }

    @Override
    public void destroy() {
        if (destroyed.compareAndSet(false, true)) {
            getDirectory().destroy();
        }
    }

    @Override
    public boolean isDestroyed() {
        return destroyed.get();
    }

    /**
     * Select a invoker using loadbalance policy.</br>
     * a) Firstly, select an invoker using loadbalance. If this invoker is in previously selected list, or,
     * if this invoker is unavailable, then continue step b (reselect), otherwise return the first selected invoker</br>
     * <p>
     * b) Reselection, the validation rule for reselection: selected > available. This rule guarantees that
     * the selected invoker has the minimum chance to be one in the previously selected list, and also
     * guarantees this invoker is available.
     *
     * @param loadbalance load balance policy
     * @param invocation  invocation
     * @param invokers    invoker candidates
     * @param selected    exclude selected invokers or not
     * @return the invoker which will final to do invoke.
     * @throws RpcException exception
     */
    protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation,
                                List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {

        if (CollectionUtils.isEmpty(invokers)) {
            return null;
        }
        // 方法名称
        String methodName = invocation == null ? StringUtils.EMPTY_STRING : RpcUtils.getMethodName(invocation);

        // 获取是否给集群容错指定一个 invoker 参数 。默认 false
        boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);

        //ignore overloaded method

        /**
         * 如果集群容错类 绑定了一个 invoker 并且 没有包含在缓存列表里面的，那么就移除掉
         */
        if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {

            stickyInvoker = null;
        }

        //ignore concurrency problem

        /**
         * 如果集群容错类 绑定了一个 invoker，并且没有使用过，那么就直接是这个 invoker，不用去 负载均衡
         */
        if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
            if (availableCheck && stickyInvoker.isAvailable()) {
                // 使用绑定的 invoker
                return stickyInvoker;
            }

        }
        // 进行负载均衡算法选择
        Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);

        /**
         * 判断是否给集群容错类 绑定一个 invoker 后续调用过程直接使用这个
         */
        if (sticky) {

            stickyInvoker = invoker;
        }

        return invoker;
    }

    private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation,
                                List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {


        if (CollectionUtils.isEmpty(invokers)) {
            return null;
        }


        if (invokers.size() == 1) {// 如果invokers=1直接获取就行了

            Invoker<T> tInvoker = invokers.get(0);
            // 校验是否有效
            checkShouldInvalidateInvoker(tInvoker);
            return tInvoker;
        }

        // 进行负载均衡
        Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

        //If the `invoker` is in the  `selected` or invoker is unavailable && availablecheck is true, reselect.
        // 判断负载均衡 选择出来的 invoker 是否存在已经使用过的 invoker 集合中
        boolean isSelected = selected != null && selected.contains(invoker);

        boolean isUnavailable = availableCheck && !invoker.isAvailable() && getUrl() != null;

        // 判断是否 不可用
        if (isUnavailable) {
            invalidateInvoker(invoker);
        }

        // 当前 负载均衡 出来的 invoker 是已经调用过了 或者 是这个 invoker 不可用了 ，那么这种情况就要重新选举 rInvoker
        if (isSelected || isUnavailable) {
            try {

                // 重新选举
                Invoker<T> rInvoker = reselect(loadbalance, invocation, invokers, selected, availableCheck);

                // 判断重新选举的是否不为 null
                if (rInvoker != null) {
                    // 不为null，那么就使用 选举 invoker
                    invoker = rInvoker;
                } else {
                    // 选举的 invoker 为 null

                    //Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
                    // 获取 上一个使用过的 invoker 索引
                    int index = invokers.indexOf(invoker);

                    try {

                        //Avoid collision
                        // 上一个使用过的 invoker 索引 +1 获取 invoker
                        invoker = invokers.get((index + 1) % invokers.size());

                    } catch (Exception e) {
                        logger.warn(CLUSTER_FAILED_RESELECT_INVOKERS,"select invokers exception","",e.getMessage() + " may because invokers list dynamic change, ignore.",e);
                    }
                }
            } catch (Throwable t) {
                logger.error(CLUSTER_FAILED_RESELECT_INVOKERS,"failed to reselect invokers","","cluster reselect fail reason is :" + t.getMessage() + " if can not solve, you can set cluster.availablecheck=false in url",t);
            }
        }

        // 最后得到负载均衡算法决策的invoker
        return invoker;
    }

    /**
     * Reselect, use invokers not in `selected` first, if all invokers are in `selected`,
     * just pick an available one using loadbalance policy.
     *
     * @param loadbalance    load balance policy
     * @param invocation     invocation
     * @param invokers       invoker candidates
     * @param selected       exclude selected invokers or not
     * @param availableCheck check invoker available if true
     * @return the reselect result to do invoke
     * @throws RpcException exception
     */
    private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
                                List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availableCheck) throws RpcException {

        // Allocating one in advance, this list is certain to be used.

        // 重新选举的 Invoker 列表
        List<Invoker<T>> reselectInvokers = new ArrayList<>(Math.min(invokers.size(), reselectCount));

        // 1. Try picking some invokers not in `selected`.
        //    1.1. If all selectable invokers' size is smaller than reselectCount, just add all
        //    1.2. If all selectable invokers' size is greater than reselectCount, randomly select reselectCount.
        //            The result size of invokers might smaller than reselectCount due to disAvailable or de-duplication (might be zero).
        //            This means there is probable that reselectInvokers is empty however all invoker list may contain available invokers.
        //            Use reselectCount can reduce retry times if invokers' size is huge, which may lead to long time hang up.


        // 重新选择数 大于 提供者 invoker 数量 没有大于默认
        if (reselectCount >= invokers.size()) {

            for (Invoker<T> invoker : invokers) {
                // check if available
                // 如果不可用 就跳过
                if (availableCheck && !invoker.isAvailable()) {
                    // add to invalidate invoker
                    invalidateInvoker(invoker);
                    continue;
                }

                // 如果该 invoker 不在已使用过的 invoker 列表 那么该 invoker 就会添加到 重新选举 Invoker 列表中 进行选举
                if (selected == null || !selected.contains(invoker)) {
                    // 添加进 重新选举 Invoker 列表
                    reselectInvokers.add(invoker);
                }
            }
        } else {
            // 如果 没有大于


            for (int i = 0; i < reselectCount; i++) {

                // select one randomly
                // 随机获取 一个
                Invoker<T> invoker = invokers.get(ThreadLocalRandom.current().nextInt(invokers.size()));


                // check if available
                // 如果不可用 就跳过
                if (availableCheck && !invoker.isAvailable()) {

                    // add to invalidate invoker
                    invalidateInvoker(invoker);
                    continue;
                }

                // de-duplication

                if (selected == null || !selected.contains(invoker) || !reselectInvokers.contains(invoker)) {
                    // 添加进 重新选举 Invoker 列表
                    reselectInvokers.add(invoker);
                }
            }

        }

        // 2. Use loadBalance to select one (all the reselectInvokers are available)

        // 重新选举 Invoker 列表 不为null，那么进行负载均衡
        if (!reselectInvokers.isEmpty()) {

            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }

        // 3. reselectInvokers is empty. Unable to find at least one available invoker.
        //    Re-check all the selected invokers. If some in the selected list are available, add to reselectInvokers.

        /**
         *  重新选举 Invoker 列表 为null，那么从已使用过的  Invoker 列表 找到可用的 Invoker 添加进 重新选举 Invoker 列表
         */
        if (selected != null) {

            for (Invoker<T> invoker : selected) {

                if ((invoker.isAvailable()) // available first

                    && !reselectInvokers.contains(invoker)) {
                    // 可用，添加进 重新选举 Invoker 列表
                    reselectInvokers.add(invoker);
                }
            }
        }

        // 4. If reselectInvokers is not empty after re-check.
        //    Pick an available invoker using loadBalance policy

        // 重新选举 Invoker 列表 不为null，那么进行负载均衡
        if (!reselectInvokers.isEmpty()) {

            return loadbalance.select(reselectInvokers, getUrl(), invocation);
        }

        // 5. No invoker match, return null.
        // 代码执行到这，证明确实是 没有可用的 invoker 了
        return null;
    }

    private void checkShouldInvalidateInvoker(Invoker<T> invoker) {
        if (availableCheck && !invoker.isAvailable()) {
            invalidateInvoker(invoker);
        }
    }

    private void invalidateInvoker(Invoker<T> invoker) {
        if (enableConnectivityValidation) {
            if (getDirectory() != null) {
                getDirectory().addInvalidateInvoker(invoker);
            }
        }
    }

    @Override
    public Result invoke(final Invocation invocation) throws RpcException {
        checkWhetherDestroyed();

        // binding attachments into invocation.
//        Map<String, Object> contextAttachments = RpcContext.getClientAttachment().getObjectAttachments();
//        if (contextAttachments != null && contextAttachments.size() != 0) {
//            ((RpcInvocation) invocation).addObjectAttachmentsIfAbsent(contextAttachments);
//        }

        InvocationProfilerUtils.enterDetailProfiler(invocation, () -> "Router route.");

        /**
         * 服务发现 获取服务提供者实例集群 （核心步骤之一）
         * 注：这里不会连接注册中心获取的，而是从本地缓存列表存获取可用 invoke 并且通过服务路由过滤掉一些 invoke
         */
        List<Invoker<T>> invokers = list(invocation);

        InvocationProfilerUtils.releaseDetailProfiler(invocation);

        // 校验 invokers 是否为 null ，为null 就抛出 RpcException 异常
        checkInvokers(invokers, invocation);

        // 获取负载均衡器
        LoadBalance loadbalance = initLoadBalance(invokers, invocation);

        // 设置Invocation 唯一id
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);

        InvocationProfilerUtils.enterDetailProfiler(invocation, () -> "Cluster " + this.getClass().getName() + " invoke.");
        try {
            /**
             * 调用集群容错 doInvoke （重点核心） 集群容错机制有如下几种：
             * 1. Failover Cluster （默认）
             *      @DubboReference(cluster = "Failover",retries = 2)
             *      失败自动切换，当出现失败，重试其它服务器。通常用于读操作，但重试会带来更长延迟。可通过 retries="2" 来设置重试次数(不含第一次)。
             * 2. Failfast Cluster
             *      快速失败，只发起一次调用，失败立即报错。通常用于非幂等性的写操作，比如新增记录。
             *
             * 3. Failsafe Cluster
             *      失败安全，出现异常时，直接忽略。通常用于写入审计日志等操作。
             * 4. Failback Cluster
             *      失败自动恢复，后台记录失败请求，定时重发。通常用于消息通知操作。
             * 5. Forking Cluster
             *      并行调用多个服务器，只要一个成功即返回。通常用于实时性要求较高的读操作，但需要浪费更多服务资源。可通过 forks="2" 来设置最大并行数
             * 6. Broadcast Cluster
             *      广播调用所有提供者，逐个调用，任意一台报错则报错。通常用于通知所有提供者更新缓存或日志等本地资源信息。
             *      现在广播调用中，可以通过 broadcast.fail.percent 配置节点调用失败的比例，当达到这个比例后，BroadcastClusterInvoker 将不再调用其他节点，直接抛出异常。
             *      broadcast.fail.percent 取值在 0～100 范围内。默认情况下当全部调用失败后，才会抛出异常。
             *      broadcast.fail.percent 只是控制的当失败后是否继续调用其他节点，并不改变结果(任意一台报错则报错)。broadcast.fail.percent 参数 在 dubbo2.7.10 及以上版本生效。
             *      Broadcast Cluster 配置 broadcast.fail.percent。
             *      broadcast.fail.percent=20 代表了当 20% 的节点调用失败就抛出异常，不再调用其他节点。
             *
             *      @DubboReference(cluster = "broadcast", parameters = {"broadcast.fail.percent", "20"})
             * 7. Available Cluster
             *      调用目前可用的实例（只调用一个），如果当前没有可用的实例，则抛出异常。通常用于不需要负载均衡的场景。
             * 8. Mergeable Cluster
             *      将集群中的调用结果聚合起来返回结果，通常和group一起配合使用。通过分组对结果进行聚合并返回聚合后的结果，比如菜单服务，用group区分同一接口的多种实现，现在消费方需从每种group中调用一次并返回结果，对结果进行合并之后返回，这样就可以实现聚合菜单项。
             * 9. ZoneAware Cluster
             *      多注册中心订阅的场景，注册中心集群间的负载均衡。对于多注册中心间的选址策略有如下四种:
             *          1. 指定优先级：preferred="true"注册中心的地址将被优先选择
             *              <dubbo:registry address="zookeeper://127.0.0.1:2181" preferred="true" />
             *          2. 同中心优先：检查当前请求所属的区域，优先选择具有相同区域的注册中心
             *              <dubbo:registry address="zookeeper://127.0.0.1:2181" zone="beijing" />
             *          3. 权重轮询：根据每个注册中心的权重分配流量
             *              <dubbo:registry id="beijing" address="zookeeper://127.0.0.1:2181" weight="100" />
             *              <dubbo:registry id="shanghai" address="zookeeper://127.0.0.1:2182" weight="10" />
             */
            return doInvoke(invocation, invokers, loadbalance);
        } finally {
            InvocationProfilerUtils.releaseDetailProfiler(invocation);
        }
    }

    protected void checkWhetherDestroyed() {
        if (destroyed.get()) {
            throw new RpcException("Rpc cluster invoker for " + getInterface() + " on consumer " + NetUtils.getLocalHost()
                + " use dubbo version " + Version.getVersion()
                + " is now destroyed! Can not invoke any more.");
        }
    }

    @Override
    public String toString() {
        return getInterface() + " -> " + getUrl().toString();
    }

    protected void checkInvokers(List<Invoker<T>> invokers, Invocation invocation) {
        if (CollectionUtils.isEmpty(invokers)) {
            throw new RpcException(RpcException.NO_INVOKER_AVAILABLE_AFTER_FILTER, "Failed to invoke the method "
                + RpcUtils.getMethodName(invocation) + " in the service " + getInterface().getName()
                + ". No provider available for the service " + getDirectory().getConsumerUrl().getServiceKey()
                + " from registry " + getDirectory()
                + " on the consumer " + NetUtils.getLocalHost()
                + " using the dubbo version " + Version.getVersion()
                + ". Please check if the providers have been started and registered.");
        }
    }

    protected Result invokeWithContext(Invoker<T> invoker, Invocation invocation) {

        Invoker<T> originInvoker = setContext(invoker);

        Result result;
        try {
            if (ProfilerSwitch.isEnableSimpleProfiler()) {
                InvocationProfilerUtils.enterProfiler(invocation, "Invoker invoke. Target Address: " + invoker.getUrl().getAddress());
            }
            setRemote(invoker, invocation);

            // 远程调用
            result = invoker.invoke(invocation);
        } finally {
            clearContext(originInvoker);

            InvocationProfilerUtils.releaseSimpleProfiler(invocation);
        }
        return result;
    }

    /**
     * Set the remoteAddress and remoteApplicationName so that filter can get them.
     *
     */
    private void setRemote(Invoker<?> invoker, Invocation invocation) {
        invocation.addInvokedInvoker(invoker);

        RpcServiceContext serviceContext = RpcContext.getServiceContext();

        serviceContext.setRemoteAddress(invoker.getUrl().toInetSocketAddress());

        serviceContext.setRemoteApplicationName(invoker.getUrl().getRemoteApplication());
    }

    /**
     * When using a thread pool to fork a child thread, ThreadLocal cannot be passed.
     * In this scenario, please use the invokeWithContextAsync method.
     *
     * @return
     */
    protected Result invokeWithContextAsync(Invoker<T> invoker, Invocation invocation, URL consumerUrl) {
        Invoker<T> originInvoker = setContext(invoker, consumerUrl);
        Result result;
        try {
            result = invoker.invoke(invocation);
        } finally {
            clearContext(originInvoker);
        }
        return result;
    }

    protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
                                       LoadBalance loadbalance) throws RpcException;

    protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
        // 获取服务发现客户端，然后获取服务提供者信息
        return getDirectory().list(invocation);
    }

    /**
     * Init LoadBalance.
     * <p>
     * if invokers is not empty, init from the first invoke's url and invocation
     * if invokes is empty, init a default LoadBalance(RandomLoadBalance)
     * </p>
     *
     * @param invokers   invokers
     * @param invocation invocation
     * @return LoadBalance instance. if not need init, return null.
     */
    protected LoadBalance initLoadBalance(List<Invoker<T>> invokers, Invocation invocation) {

        ApplicationModel applicationModel = ScopeModelUtil.getApplicationModel(invocation.getModuleModel());
        if (CollectionUtils.isNotEmpty(invokers)) {

            // 获取负载均衡器
            return applicationModel.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl().getMethodParameter(
                    // 如果没有指定使用那个负载均衡 那么默认使用随机
                    RpcUtils.getMethodName(invocation), LOADBALANCE_KEY, DEFAULT_LOADBALANCE
                )
            );
        } else {
            return applicationModel.getExtensionLoader(LoadBalance.class).getExtension(DEFAULT_LOADBALANCE);
        }
    }


    private Invoker<T> setContext(Invoker<T> invoker) {
        return setContext(invoker, null);
    }

    private Invoker<T> setContext(Invoker<T> invoker, URL consumerUrl) {
        RpcServiceContext context = RpcContext.getServiceContext();
        Invoker<?> originInvoker = context.getInvoker();
        context.setInvoker(invoker).setConsumerUrl(null != consumerUrl ? consumerUrl : RpcContext.getServiceContext().getConsumerUrl());
        return (Invoker<T>) originInvoker;
    }

    private void clearContext(Invoker<T> invoker) {
        // do nothing
        RpcContext context = RpcContext.getServiceContext();
        context.setInvoker(invoker);
    }
}
